package com.bigdata;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.MD5Hash;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class HBaseDMLDemo {

    private Configuration conf;
    // HBase 的连接对象
    private Connection conn;
    // HBase 操作DDL的对象
    private Admin admin;

    @Before
    public void init() throws Exception {
        conf = HBaseConfiguration.create();
        // 设置zk的连接信息
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }

    /**
     * scan 扫描一批数据
     */
    @Test
    public void scanTest() throws Exception {
        // scan 'api_test:student'
        TableName tableName = TableName.valueOf("MYTABLE");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        Scan scan = new Scan();

        // 追加扫描条件
//        scan.withStartRow("1500100011".getBytes());
//        scan.withStopRow("1500100018".getBytes());

        // 发送scan命令，获取扫描结果
        ResultScanner scanner = table.getScanner(scan);
        Result result;
        while ((result = scanner.next()) != null) {
            String rowkey = new String(result.getRow());
            System.out.println(rowkey);
            Cell[] cells = result.rawCells();
            result.getRow();
            for (Cell cell : cells) {
                String msg = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(msg);
            }
        }

        // 关闭资源
        table.close();
    }


    /**
     * 删除数据
     */
    @Test
    public void deleteTest() throws Exception {
        // delete 't1', 'r1', 'c1'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        Delete delete = new Delete("1500100005".getBytes());
        delete.addColumn("info".getBytes(), "age".getBytes());

        table.delete(delete);

        table.close();
    }

    /**
     * get一条数据
     */
    @Test
    public void getCountTest() throws Exception {
        // put 'api_test:student', '1001', 'info:name', 'zs'
        TableName tableName = TableName.valueOf("default:test4");


        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        // get 'api_test:student', '1500100994', {COLUMN => 'info:age'}
        Get get = new Get("1".getBytes());


        Result result = table.get(get);

        String rowkey = new String(result.getRow());
        System.out.println(rowkey);
        Cell[] cells = result.rawCells();
        result.getRow();
        for (Cell cell : cells) {
            String msg = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println(msg);
        }


        // 关闭资源
        table.close();
    }


    /**
     * get一条数据
     */
    @Test
    public void getDataTest() throws Exception {
        // put 'api_test:student', '1001', 'info:name', 'zs'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        // get 'api_test:student', '1500100994', {COLUMN => 'info:age'}
        Get get = new Get("1500100994".getBytes());
        Get get1 = new Get("1500100005".getBytes());
        // 指定get的列
        get.addColumn("info".getBytes(), "age".getBytes());
        get.addColumn("info".getBytes(), "name".getBytes());


        Result[] results = table.get(Arrays.asList(get, get1));

        Result result = results[1];

//        String rowkey = new String(result.getRow());
//        System.out.println(rowkey);
//        Cell[] cells = result.rawCells();
//        result.getRow();
//        for (Cell cell : cells) {
//            String msg = Bytes.toString(CellUtil.cloneValue(cell));
//            System.out.println(msg);
//        }

        processResult(result);

        // 关闭资源
        table.close();
    }


    /**
     * 打印扫描结果
     *
     * @param scanner
     */
    private static void processScanner(ResultScanner scanner) {
        try {
            Result result;
            int count = 0;
            while ((result = scanner.next()) != null) {
                processResult(result);
                count++;
            }
            System.out.println("此次扫描到了 " + count + " 行数据");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 打印一个result对象中的数据
     *
     * @param result
     */
    private static void processResult(Result result) {
        // 获取rowkey
        String rowkey = Bytes.toString(result.getRow());

//        String name = new String(result.getValue("info".getBytes(), "name".getBytes()));
        // 列不存在，不会出现空指针
        String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
        String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
        String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));

        System.out.println("id:" + rowkey + "---name:" + name + "---age:" + age
                + "---gender:" + gender + "---clazz:" + clazz);
    }


    /**
     * 批量保存数据到表中
     */
    @Test
    public void saveDataBatchTest() throws Exception {
        // put 'api_test:student', '1001', 'info:name', 'zs'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("data/students.txt")));


        List<Put> puts = new ArrayList<>();
        String line;
        while ((line = br.readLine()) != null) {
            System.out.println(line);
            String[] splits = line.split(",");

            Put put = new Put(splits[0].getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), splits[1].getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), splits[2].getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), splits[3].getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), splits[4].getBytes());
            puts.add(put);
        }

        // 批量添加
        table.put(puts);

        // 关闭资源
        table.close();


    }

    /**
     * 保存数据到表中
     */
    @Test
    public void saveDataTest() throws Exception {
        // put 'api_test:student', '1001', 'info:name', 'zs'
        TableName tableName = TableName.valueOf("api_test:student");

        // 和DML相关的操作，需要使用Table对象
        Table table = conn.getTable(tableName);

        Put put = new Put("1001".getBytes());

        put.addColumn("info".getBytes(), "name".getBytes(), "zs".getBytes());

        table.put(put);

        // 关闭资源
        table.close();
    }

    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }
}
